package com.moli.iov.handle.alarm.mq.kafka;

import com.alibaba.fastjson.JSONObject;
import com.moli.iov.consumer.KafkaConsumerServer;
import com.moli.iov.handle.alarm.biz.OfflineHandler;
import com.moli.iov.handle.alarm.enums.OnOffTypeEnum;
import com.moli.iov.handle.alarm.model.dto.OnOffLineAlarmDto;
import com.moli.iov.protocol.model.dto.VehicleOnLineDto;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

/**
 * <p>
 *  车辆在线、离线报警
 * </p>
 *
 * @author: zql
 * @Date: 2019/2/19 14:06
 */
@Slf4j
public class OnOffConsumerMessageListener extends KafkaConsumerServer {

    @Autowired
    private ThreadPoolTaskExecutor taskExecutor;

    @Autowired
    private OfflineHandler offlineHandler;

    @Override
    public void manualHandle(ConsumerRecord<String, Object> record,
                             Acknowledgment acknowledgment) {
        try{

            onlineHandle(record.topic(),
                    (VehicleOnLineDto)record.value());

        }catch(Exception e){
            log.error("kafka 消费者 报警 车辆在线恢复 车辆离线报警取消 添加线程池异常", e);
        }finally{
            acknowledgment.acknowledge();
        }
    }

    private void onlineHandle(String topic,
                              VehicleOnLineDto vehicleOnLineDto){
        taskExecutor.execute(() -> {

            try {
                if(vehicleOnLineDto == null){
                    return;
                }
                log.info("Kafka 消费者 报警 车辆在线恢复 车辆离线报警取消 线程池中处理开始 接收到消息  topic: {} 数据：{}",
                        topic, JSONObject.toJSON(vehicleOnLineDto));

                if(vehicleOnLineDto == null){
                    log.error("Kafka 消费者 报警 车辆在线恢复 车辆离线报警取消 无报警信息 topic: {} 数据：{}",
                            topic, vehicleOnLineDto);
                    return;
                }

                offlineHandler.handle(new OnOffLineAlarmDto(vehicleOnLineDto,
                                                            OnOffTypeEnum.REPORTING.getFlag()));
            }catch(Exception e){
                log.error("aKafka 消费者 报警 车辆在线恢复 车辆离线报警取消  处理异常", e);
            }
        });
    }
}
